package rdd.operate;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Serializable;

import java.util.Arrays;
import java.util.List;

public class Spark70_Operate_Serializable {
    public static void main(String[] args) {
        final SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName("spark");
        final JavaSparkContext jsc = new JavaSparkContext(conf);

        final List<String> nums = Arrays.asList("Hadoop","Hive","Flink","Spark");
        JavaRDD<String> rdd = jsc.parallelize(nums, 2);
        Search search = new Search("H");
        search.match(rdd);
    }
}

class Search implements Serializable {
    private String query;
    public Search(String query){
        this.query = query;
    }

    public void match( JavaRDD<String> rdd) {
       /*
       如果代码是这样，就不需要实现序列化接口，因为this.query赋值给q实在Driver端实现的，并且String传输可以自行序列化
       并且Search创建在堆内存中，String是与方法共存，创建在栈内存中
        String q = this.query;
       rdd.filter(
               s -> s.startsWith(q)
       ).collect().forEach(System.out::println);
       */
       rdd.filter(
                //这里需要传入query是该对象的属性，对象实在Driver端创建的，属于Driver端
                //但是在rdd算子都是在Executor端进行的，但是在executor端需要使用对象属性，就要把属性传到exector端，通过网络传输
                //网络不能传输对象，必须对对象进行序列化，实现序列化接口
                s -> s.startsWith(this.query)
        ).collect().forEach(System.out::println);
    }
}
